package com.org.oracle.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.RandomUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.org.oracle.constant.MQConstants;
import com.org.oracle.constant.RedisKeyConstants;
import com.org.oracle.enums.FollowLuaResultEnum;
import com.org.oracle.enums.ResponseCodeEnum;
import com.org.oracle.enums.UnfollowLuaResultEnum;
import com.org.oracle.exception.CustomException;
import com.org.oracle.mapper.FollowingMapper;
import com.org.oracle.mongodb.vo.UserSimpleVO;
import com.org.oracle.mysql.dto.FollowUserDTO;
import com.org.oracle.mysql.dto.FollowUserMqDTO;
import com.org.oracle.mysql.dto.UnfollowUserDTO;
import com.org.oracle.mysql.dto.UnfollowUserMqDTO;
import com.org.oracle.mysql.entity.Following;
import com.org.oracle.service.UserRelationService;
import com.org.oracle.service.UserService;
import com.org.oracle.util.DateUtils;
import com.org.oracle.util.JsonUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * @author: heiye
 * @date: 2024/12/20 上午9:14
 * @version: v1.0.0
 * @description: 用户关系
 */
@Slf4j
@Service
public class UserRelationServiceImpl implements UserRelationService {

    @Resource
    private UserService userService;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private FollowingMapper followingMapper;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Resource(name = "userRelationExecutor")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    /**
     * 关注用户
     *
     * @param followUserDTO
     */
    @Override
    public void follow(String uid, FollowUserDTO followUserDTO) {
        // 关注的用户 ID
        String followUserId = followUserDTO.getFollowUserId();

        // 校验：无法关注自己
        if (StringUtils.equals(uid, followUserId)) {
            throw new CustomException(ResponseCodeEnum.CANT_FOLLOW_YOUR_SELF);
        }

        // 校验：关注的用户是否存在
        UserSimpleVO userSimpleVO = userService.findUserSimpleById(followUserId);
        if (Objects.isNull(userSimpleVO)) {
            throw new CustomException(ResponseCodeEnum.FOLLOW_USER_NOT_EXISTED);
        }

        // 构建当前用户关注列表的 Redis Key
        String followingKey = RedisKeyConstants.buildUserFollowingKey(uid);

        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        // Lua 脚本路径
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/follow_check_and_add.lua")));
        // 返回类型
        script.setResultType(Long.class);

        // 当前时间
        LocalDateTime now = LocalDateTime.now();
        // 当前时间转时间戳
        long timestamp = DateUtils.localDateTime2Timestamp(now);

        // 执行 Lua脚本
        Long result = redisTemplate.execute(script, Collections.singletonList(followingKey), followUserId, timestamp);

        // 校验 Lua 脚本执行结果
        checkFollowingLuaScriptResult(result);

        // ZSET 不存在
        if (Objects.equals(result, FollowLuaResultEnum.ZSET_NOT_EXIST.getCode())) {
            // 从数据库查询当前用户的关注关系记录
            List<Following> followings = selectFollowingByUserId(uid);

            // 随机过期时间
            // 保底1天+随机秒数
            long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);

            // 若记录为空，直接 ZADD 关系数据, 并设置过期时间
            if (CollUtil.isEmpty(followings)) {
                DefaultRedisScript<Long> script2 = new DefaultRedisScript<>();
                script2.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/follow_add_and_expire.lua")));
                script2.setResultType(Long.class);
                redisTemplate.execute(script2, Collections.singletonList(followingKey), followUserId, DateUtils.localDateTime2Timestamp(now), expireSeconds);
            }
            // 若记录不为空，则将关注关系数据全量同步到 Redis 中，并设置过期时间；
            else {
                // ZSET 批量添加用户关注列表
                followBatchAddAndExpire(followings, expireSeconds, followingKey);

                // 再次调用上面的 Lua 脚本：follow_check_and_add.lua , 将最新的关注关系添加进去
                result = redisTemplate.execute(script, Collections.singletonList(followingKey), followUserId, timestamp);

                checkFollowingLuaScriptResult(result);
            }
        }

        // 发送 MQ
        // 构建消息体
        FollowUserMqDTO followUserMqDTO = FollowUserMqDTO.builder()
                .userId(uid)
                .followingUserId(followUserId)
                .createTime(now)
                .build();

        // 构建消息对象，并将 DTO 转成 Json 字符串设置到消息体中
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(followUserMqDTO)).build();

        // 通过冒号连接, 可让 MQ 发送给主题 Topic 时，携带上标签 Tag
        String destination = MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW + ":" + MQConstants.TAG_FOLLOW;

        log.info("==> 开始发送关注操作 MQ, 消息体: {}", followUserMqDTO);

        // 异步发送 MQ 消息，提升接口响应速度
        rocketMQTemplate.asyncSendOrderly(destination, message, uid, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【关注用户】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【关注用户】MQ 发送失败", throwable);
            }
        });
    }

    /**
     * 取消关注用户
     *
     * @param uid
     * @param unFollowUserDTO
     */
    @Override
    public void unfollow(String uid, UnfollowUserDTO unFollowUserDTO) {
        // 取关用户的 ID
        String unfollowUserId = unFollowUserDTO.getUnfollowUserId();

        // 校验：无法取关自己
        if (StringUtils.equals(uid, unfollowUserId)) {
            throw new CustomException(ResponseCodeEnum.CANT_UNFOLLOW_YOUR_SELF);
        }

        // 校验：关注的用户是否存在
        UserSimpleVO userSimpleVO = userService.findUserSimpleById(unfollowUserId);
        if (Objects.isNull(userSimpleVO)) {
            throw new CustomException(ResponseCodeEnum.FOLLOW_USER_NOT_EXISTED);
        }

        // 当前用户的关注列表 Redis Key
        String followingKey = RedisKeyConstants.buildUserFollowingKey(uid);

        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        // Lua 脚本路径
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/unfollow_check_and_delete.lua")));
        // 返回类型
        script.setResultType(Long.class);
        // 执行 Lua 脚本
        Long result = redisTemplate.execute(script, Collections.singletonList(followingKey), unfollowUserId);

        // 校验 Lua 脚本执行结果
        // 取关的用户不在关注列表中
        if (Objects.equals(result, UnfollowLuaResultEnum.NOT_FOLLOWED.getCode())) {
            throw new CustomException(ResponseCodeEnum.FOLLOW_USER_NOT_EXISTED);
        }

        // ZSET 关注列表不存在
        if (Objects.equals(result, UnfollowLuaResultEnum.ZSET_NOT_EXIST.getCode())) {
            // 从数据库查询当前用户的关注关系记录
            List<Following> followings = selectFollowingByUserId(uid);

            // 若记录为空，则表示还未关注任何人，提示还未关注对方
            if (CollUtil.isEmpty(followings)) {
                throw new CustomException(ResponseCodeEnum.NOT_FOLLOWED);
            }
            // 若记录不为空，则将关注关系数据全量同步到 Redis 中，并设置过期时间；
            else {
                // 随机过期时间
                // 保底1天+随机秒数
                long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);

                // ZSET 批量添加用户关注列表
                followBatchAddAndExpire(followings, expireSeconds, followingKey);

                // 再次调用上面的 Lua 脚本：unfollow_check_and_delete.lua , 将取关的用户删除
                result = redisTemplate.execute(script, Collections.singletonList(followingKey), unfollowUserId);
                // 再次校验结果
                if (Objects.equals(result, UnfollowLuaResultEnum.NOT_FOLLOWED.getCode())) {
                    throw new CustomException(ResponseCodeEnum.FOLLOW_USER_NOT_EXISTED);
                }
            }
        }

        // 发送 MQ
        // 构建消息体 DTO
        UnfollowUserMqDTO unfollowUserMqDTO = UnfollowUserMqDTO.builder()
                .userId(uid)
                .unfollowingUserId(unfollowUserId)
                .createTime(LocalDateTime.now())
                .build();

        // 构建消息对象，并将 DTO 转成 Json 字符串设置到消息体中
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(unfollowUserMqDTO)).build();

        // 通过冒号连接, 可让 MQ 发送给主题 Topic 时，携带上标签 Tag
        String destination = MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW + ":" + MQConstants.TAG_UNFOLLOW;

        // 异步发送 MQ 消息，提升接口响应速度
        rocketMQTemplate.asyncSendOrderly(destination, message, uid, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【取关用户】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("==> 【取关用户】MQ 发送失败", throwable);
            }
        });
    }

    /**
     * 查询用户之间关系
     *
     * @param uid
     * @param targetUserId
     */
    @Override
    public Boolean findUserRelation(String uid, String targetUserId) {
        // 校验：自己不会关注自己
        if (Objects.equals(uid, targetUserId)) {
            return false;
        }
        // 先在 Redis 中查询用户关系列表
        String followingKey = RedisKeyConstants.buildUserFollowingKey(uid);

        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        // Lua 脚本路径
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/follow_check.lua")));
        // 返回类型
        script.setResultType(Long.class);
        // 执行 Lua 脚本
        Long result = redisTemplate.execute(script, Collections.singletonList(followingKey), targetUserId);

        FollowLuaResultEnum followLuaResultEnum = FollowLuaResultEnum.valueOf(result);

        // 返回类型错误
        if (Objects.isNull(followLuaResultEnum)) {
            throw new RuntimeException("Lua 返回结果错误");
        }

        // Redis 不存在该用户关系列表
        if (Objects.equals(result, FollowLuaResultEnum.ZSET_NOT_EXIST.getCode())) {
            // 数据库中查询用户关系
            List<Following> followings = selectFollowingByUserId(uid);

            // 随机过期时间
            // 保底1天+随机秒数
            long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);

            // 用户关注列表不为空同步到 redis 中
            if (CollUtil.isNotEmpty(followings)) {
                // 执行 Lua 脚本，批量同步关注关系数据到 Redis 中
                followBatchAddAndExpire(followings, expireSeconds, followingKey);
                // 再次调用上面的 Lua 脚本判断是否存在关注
                result = redisTemplate.execute(script, Collections.singletonList(followingKey), targetUserId);
                // 返回结果
                return Objects.equals(result, FollowLuaResultEnum.ALREADY_FOLLOWED.getCode());
            }

            // 用户关注列表为空，防止缓存穿透
            threadPoolTaskExecutor.submit(()-> {
                // 异步提交到 redis
                redisTemplate.opsForZSet().add(followingKey, "null", 0);
                redisTemplate.expire(followingKey, expireSeconds, TimeUnit.SECONDS);
            });
            return false;
        }

        // 判断目标用户是否存在关注列表中
        return Objects.equals(result, FollowLuaResultEnum.ALREADY_FOLLOWED.getCode());
    }

    /**
     * 执行 Lua 脚本，批量同步关注关系数据到 Redis 中
     *
     * @param followings
     * @param expireSeconds
     * @param followingKey
     */
    private void followBatchAddAndExpire(List<Following> followings, long expireSeconds, String followingKey) {
        // 构建 Lua 参数
        Object[] luaArgs = buildLuaArgs(followings, expireSeconds);

        // 执行 Lua 脚本，批量同步关注关系数据到 Redis 中
        DefaultRedisScript<Long> script3 = new DefaultRedisScript<>();
        // Lua 脚本路径
        script3.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/follow_batch_add_and_expire.lua")));
        // 返回类型
        script3.setResultType(Long.class);

        // TODO: 可以根据用户类型，设置不同的过期时间，若当前用户为大V, 则可以过期时间设置的长些或者不设置过期时间；如不是，则设置的短些
        // 如何判断呢？可以从计数服务获取用户的粉丝数，目前计数服务还没创建，则暂时采用统一的过期策略
        redisTemplate.execute(script3, Collections.singletonList(followingKey), luaArgs);
    }

    /**
     * 从数据库查询当前用户的关注关系记录
     *
     * @param uid
     * @return
     */
    private List<Following> selectFollowingByUserId(String uid) {
        // 从数据库查询当前用户的关注关系记录
        LambdaQueryWrapper<Following> lambdaQueryWrapper = Wrappers.<Following>lambdaQuery()
                .select(Following::getUserId, Following::getFollowingUserId, Following::getCreateTime)
                .eq(Following::getUserId, uid);
        return followingMapper.selectList(lambdaQueryWrapper);
    }

    /**
     * 校验 Lua 脚本结果，根据状态码抛出对应的业务异常
     *
     * @param result
     */
    private static void checkFollowingLuaScriptResult(Long result) {
        FollowLuaResultEnum followLuaResultEnum = FollowLuaResultEnum.valueOf(result);
        // 返回类型错误
        if (Objects.isNull(followLuaResultEnum)) {
            throw new RuntimeException("Lua 返回结果错误");
        }

        switch (followLuaResultEnum) {
            // 已经关注了该用户
            case ALREADY_FOLLOWED -> throw new CustomException(ResponseCodeEnum.ALREADY_FOLLOWED);
            // 关注数已达到上限
            case FOLLOW_LIMIT -> throw new CustomException(ResponseCodeEnum.FOLLOWING_COUNT_LIMIT);
        }
    }

    /**
     * 构建 Lua 脚本参数
     *
     * @param followings
     * @param expireSeconds
     * @return
     */
    private static Object[] buildLuaArgs(List<Following> followings, long expireSeconds) {
        // 每个关注关系有 2 个参数（score 和 value），再加一个过期时间
        int argsLength = followings.size() * 2 + 1;
        Object[] luaArgs = new Object[argsLength];

        int i = 0;
        for (Following following : followings) {
            // 关注时间作为 score
            luaArgs[i] = DateUtils.localDateTime2Timestamp(following.getCreateTime());
            // 关注的用户 ID 作为 ZSet value
            luaArgs[i + 1] = following.getFollowingUserId();
            i += 2;
        }
        // 最后一个参数是 ZSet 的过期时间
        luaArgs[argsLength - 1] = expireSeconds;
        return luaArgs;
    }
}
